package com.ccsoft.thread.service;

import ch.qos.logback.core.pattern.FormatInfo;
import com.sun.media.jfxmedia.logging.Logger;
import lombok.extern.slf4j.Slf4j;
import sun.font.FontManagerNativeLibrary;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 2021年12月20日 13:35:12 chanchaw
 * 自定义线程池
 */
@Slf4j
public class No17ThreadPool {

    public static void main(String[] args) {
        ThreadPool tp = new ThreadPool(1,1000,TimeUnit.MILLISECONDS,1,(queue,task)->{
            //queue.put(task); 1. 任务队列满了死等
            //queue.put(task,1500,TimeUnit.MILLISECONDS);// 2. 超时等待
            // 3. 不做任何操作即放弃了队列满之后的任务
            // 4. 抛出异常不再执行任何任务
//            throw new RuntimeException("没有执行完毕的任务都会由于异常而不执行");

            // 5. 主线程自己执行
            task.run();
        });
        for (int i = 0; i < 3; i++) {
            int j = i;
            tp.execute(()->{
                try {
                    TimeUnit.MILLISECONDS.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.info("分配给线程池任务{}",j);
            });
        }
    }
}

@Slf4j
class ThreadPool {
    private HashSet<Worker> workers = new HashSet<>();
    private BlockingQueue<Runnable> taskQueue;

    private int size;
    private long timeout;// 任务超时时间
    private TimeUnit timeUnit;
    private RejectPolicy<Runnable> rejectPolicy;

    public ThreadPool(int size, long timeout, TimeUnit timeUnit, int taskQueueSize,RejectPolicy<Runnable> rejectPolicy) {
        this.size = size;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        taskQueue = new BlockingQueue<>(taskQueueSize);
        this.rejectPolicy = rejectPolicy;
    }

    public void execute(Runnable task){
        synchronized (this){
            if(this.workers.size() < this.size){
                Worker worker = new Worker(task);
                log.info("创建了worker:{},task:{}",worker,task);
                workers.add(worker);
                worker.start();
            }else{
                log.info("进入队列的任务：{}",task);
//                taskQueue.put(task);
                taskQueue.tryPut(rejectPolicy,task);
            }
        }
    }

    class Worker extends Thread {
        private Runnable task;
        public Worker(Runnable task){
            this.task = task;
        }

        @Override
        public void run() {
            // worker 执行完当前 task 再到任务队列 taskQueue 中查找是否有等待的任务
            // 有的话则取出并执行任务。从任务队列取任务有等待时间，超过该时间则
            // 线程不再等待直接返回 NULL 任务
            while(this.task != null || (task = taskQueue.take(timeout, timeUnit)) != null){
                try {
                    log.info("执行了任务：{}",task);
                    task.run();
                }catch(Exception e){
                    e.printStackTrace();
                }finally {
                    task = null;
                }
            }

            // 当任务队列的任务也都执行完毕了就删除该 worker
            synchronized (workers){
                log.info("移除了任务：{}",task);
                workers.remove(this);
            }
        }
    }
}

// 阻塞队列。初始化队列时设置队列任务数量的上限
// 工作线程获取任务时如果队列为空则阻塞，等待有新任务被填充进去
// 生产者线程填充任务时如果队列已满则阻塞，等待有消费者线程消费任务后被唤醒
// 因为有两个条件变量：队列空、队列满，所以选择使用 ReentrantLock
class BlockingQueue<T> {
    private Deque<T> deque = new ArrayDeque<>();
    private int size;// 队列任务数量的上限，达到该数量之后不再追加
    private ReentrantLock lock = new ReentrantLock();

    // 队列已满的条件变量。此时再有任务准备添加进入队列则将线程纳入该条件变量
    private Condition fullSet = lock.newCondition();
    // 队列空的条件变量。当队列空时仍有 worker 从队列获取任务则进入该条件变量
    private Condition emptySet = lock.newCondition();

    public BlockingQueue(int size) {
        this.size = size;
    }

    // 向队列填充任务，没有时间限制，如果队列已经满了，则无限制等待下去
    public void put(T ele){
        lock.lock();
        try {
            while( this.deque.size() >= this.size ) {
                try {
                    fullSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            this.deque.addLast(ele);
            emptySet.signal();// 想队列中添加任务后唤醒空条件中的线程
        }finally {
            lock.unlock();
        }
    }

    // 有超时时间的进入队列，
    public boolean put(T ele,long timeout, TimeUnit timeUnit){
        lock.lock();
        try {
            long nanos = timeUnit.toNanos(timeout);
            while( this.deque.size() >= this.size ) {
                try {
                    if(nanos <= 0) return false;// 超时没有加入队列则返回 false
                    nanos = fullSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            this.deque.addLast(ele);
            emptySet.signal();// 想队列中添加任务后唤醒空条件中的线程
            return true;
        }finally {
            lock.unlock();
        }
    }

    // 从队列中获取任务
    public T take(){
        lock.lock();
        try {
            while( this.deque.isEmpty() ){
                try {
                    emptySet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            T ele = this.deque.removeFirst();
            fullSet.signal();// 移除一个任务后唤醒满条件
            return ele;
        } finally {
            lock.unlock();
        }
    }

    // 从队列获取任务时有超时时间，超过该时间则不再获取
    public T take(long timeout, TimeUnit timeUnit){
        lock.lock();
        try {
            long surplus = timeUnit.toNanos(timeout);
            while( this.deque.isEmpty() ){
                try {
                    if( surplus<=0 ) return null;// 等待时间结束了队列仍然是空则返回空
                    surplus = emptySet.awaitNanos(surplus);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            T ele = this.deque.removeFirst();
            fullSet.signal();// 移除一个任务后唤醒满条件
            return ele;
        } finally {
            lock.unlock();
        }
    }

    // 返回当前队列中的任务数量，多线程中保证返回数量的时没有任务进出
    // 否则返回的数量不准确
    public int getSize(){
        lock.lock();
        try {
            return this.deque.size();
        }finally {
            lock.unlock();
        }
    }

    public void tryPut(RejectPolicy<T> rejectPolicy, T task) {
        lock.lock();
        try {
            if(deque.size() < size){
                deque.addLast(task);
                emptySet.signal();
            }else{
                rejectPolicy.reject(this,task);
            }
        }finally {
            lock.unlock();
        }
    }
}

// 队列满时追加任务的拒绝策略
@FunctionalInterface
interface RejectPolicy<T> {
    void reject(BlockingQueue<T> queue,T task);
}
